package com.iflytek.controller;

import com.iflytek.bean.AreaControl;
import com.iflytek.bean.MonitorInfo;
import com.iflytek.schema.JSONDeserializationSchema;
import com.iflytek.utils.BloomFilterUtil;
import com.iflytek.utils.Constants;
import com.iflytek.utils.JedisUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.util.Properties;


public class _06VehicleDistributionV3 {

    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);  //设置并行度

        //2.设置数据源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","hadoop102:9092");
        properties.setProperty("group.id","g5");

        FlinkKafkaConsumer<MonitorInfo> consumer = new FlinkKafkaConsumer<MonitorInfo>("topic-car",
                new JSONDeserializationSchema<>(MonitorInfo.class),properties);
        DataStreamSource<MonitorInfo> ds1 = env.addSource(consumer);
        SingleOutputStreamOperator<AreaControl> resultDf = ds1.keyBy(k -> k.getAreaId())
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
                .apply(new WindowFunction<MonitorInfo, AreaControl, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<MonitorInfo> input, Collector<AreaControl> out) throws Exception {

                        // redis中的bitmaps 跟 布隆过滤器 有啥关系？
                        // 布隆过滤器是一种数据结构，可以理解为算法，算法落地--> bitmaps
                        Jedis jedis = JedisUtils.getJedis();
                        String start = DateFormatUtils.format(window.getStart(), "yyyy-MM-dd HH:mm");
                        String end = DateFormatUtils.format(window.getEnd(), "yyyy-MM-dd HH:mm");
                        String  redisKey = "";
                        int carCount = 0;
                        for (MonitorInfo monitorInfo : input) {
                            redisKey = "area:"+key+":"+start;
                            int[] offsets = BloomFilterUtil.getOffsets(monitorInfo.getCar());
                            boolean r1 = jedis.getbit(redisKey, offsets[0]);
                            boolean r2 = jedis.getbit(redisKey, offsets[1]);
                            if(r1 == false || r2 == false){
                                carCount +=1;
                            }

                            jedis.setbit(redisKey,offsets[0],true);
                            jedis.setbit(redisKey,offsets[1],true);
                        }

                        // 每一个区域 近10分钟之内有多少量车
                        out.collect(new AreaControl(null, key, carCount, start, end));
                    }
                });
        resultDf.addSink(JdbcSink.sink(
                "insert into t_area_control values(null,?,?,?,?)",
                (PreparedStatement ps, AreaControl areaControl) -> {
                    ps.setString(1, areaControl.getAreaId());
                    ps.setInt(2, areaControl.getCarCount());
                    ps.setString(3, areaControl.getWindowStart());
                    ps.setString(4, areaControl.getWindowEnd());
                },
                JdbcExecutionOptions.builder().withBatchSize(1).withBatchIntervalMs(5000).build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(Constants.URL)
                        .withDriverName(Constants.DRIVER)
                        .withUsername(Constants.USERNAME)
                        .withPassword(Constants.PASSWORD)
                        .build()));

        env.execute();
    }

}
